
package org.apache.seatunnel.connectors.seatunnel.gaussdbcassandra.config;

import com.datastax.oss.driver.api.core.ConsistencyLevel;
import com.datastax.oss.driver.api.core.DefaultConsistencyLevel;
import com.datastax.oss.driver.api.core.cql.DefaultBatchType;
import lombok.Getter;
import lombok.Setter;
import org.apache.seatunnel.shade.com.typesafe.config.Config;

import java.io.Serializable;
import java.util.List;

@Setter
@Getter
public class GaussDBCassandraParameters implements Serializable {
    private static final long serialVersionUID = 1566461332660097443L;

    private String host;
    private String username;
    private String password;
    private String datacenter;
    private String keyspace;
    private String table;
    private String cql;
    private List<String> fields;
    private ConsistencyLevel consistencyLevel;
    private Integer batchSize;
    private DefaultBatchType batchType;
    private Boolean asyncWrite;

    public void buildWithConfig(Config config) {

        this.host = config.getString(GaussDBCassandraConfig.HOST.key());
        this.keyspace = config.getString(GaussDBCassandraConfig.KEYSPACE.key());

        if (config.hasPath(GaussDBCassandraConfig.USERNAME.key())) {
            this.username = config.getString(GaussDBCassandraConfig.USERNAME.key());
        }
        if (config.hasPath(GaussDBCassandraConfig.PASSWORD.key())) {
            this.password = config.getString(GaussDBCassandraConfig.PASSWORD.key());
        }
        if (config.hasPath(GaussDBCassandraConfig.DATACENTER.key())) {
            this.datacenter = config.getString(GaussDBCassandraConfig.DATACENTER.key());
        } else {
            this.datacenter = GaussDBCassandraConfig.DATACENTER.defaultValue();
        }
        if (config.hasPath(GaussDBCassandraConfig.TABLE.key())) {
            this.table = config.getString(GaussDBCassandraConfig.TABLE.key());
        }
        if (config.hasPath(GaussDBCassandraConfig.CQL.key())) {
            this.cql = config.getString(GaussDBCassandraConfig.CQL.key());
        }
        if (config.hasPath(GaussDBCassandraConfig.FIELDS.key())) {
            this.fields = config.getStringList(GaussDBCassandraConfig.FIELDS.key());
        }
        if (config.hasPath(GaussDBCassandraConfig.CONSISTENCY_LEVEL.key())) {
            this.consistencyLevel = DefaultConsistencyLevel
                    .valueOf(config.getString(GaussDBCassandraConfig.CONSISTENCY_LEVEL.key()));
        } else {
            this.consistencyLevel = DefaultConsistencyLevel
                    .valueOf(GaussDBCassandraConfig.CONSISTENCY_LEVEL.defaultValue());
        }

        if (config.hasPath(GaussDBCassandraConfig.BATCH_SIZE.key())) {
            this.batchSize = config.getInt(GaussDBCassandraConfig.BATCH_SIZE.key());
        } else {
            this.batchSize = GaussDBCassandraConfig.BATCH_SIZE.defaultValue();
        }

        if (config.hasPath(GaussDBCassandraConfig.BATCH_TYPE.key())) {
            this.batchType = DefaultBatchType.valueOf(config.getString(GaussDBCassandraConfig.BATCH_TYPE.key()));
        } else {
            this.batchType = DefaultBatchType.valueOf(GaussDBCassandraConfig.BATCH_TYPE.defaultValue());
        }

        if (config.hasPath(GaussDBCassandraConfig.ASYNC_WRITE.key())) {
            this.asyncWrite = config.getBoolean(GaussDBCassandraConfig.ASYNC_WRITE.key());
        } else {
            this.asyncWrite = GaussDBCassandraConfig.ASYNC_WRITE.defaultValue();
        }
    }
}
